package com.ibm.cps.processors;

import java.util.Arrays;
import java.util.Collection;

import com.ibm.cps.message.ParsedDataMessage;
import com.ibm.cps.messagequeue.AMQPClient;
import com.ibm.cps.sql.expression.function.meta.IgnoredInFunctionCall;
import com.ibm.cps.sql.expression.function.meta.ValueParam;
import com.ibm.datamodel.timeseries.SchemaFormat;
import com.ibm.factories.MessageParserFactory;
import com.ibm.interfaces.IMessageClient;
import com.ibm.interfaces.IMessageParser;
import com.ibm.interfaces.ITimeSeriesSchema;
import com.ibm.util.ErrorCode;
import com.ibm.util.exception.CPSException;

/*
 * use outputProcessor instead
 */
@Deprecated
public class PublishProcessor extends BasicProcessor {

	private ITimeSeriesSchema parentSchema;
	private IMessageClient publisher;
	
	@ValueParam @IgnoredInFunctionCall
	private String topicName;
	@ValueParam @IgnoredInFunctionCall
	private String server;
	@ValueParam @IgnoredInFunctionCall
	private int port;
	@ValueParam @IgnoredInFunctionCall
	private String clientName;
	@ValueParam @IgnoredInFunctionCall
	private String password;
	private IMessageParser parser;
	

	public PublishProcessor(String tenantId, String processorId, String source, ITimeSeriesSchema parentSchema) throws CPSException {
		super(tenantId, processorId, source, parentSchema);
		this.parentSchema = parentSchema;
	}
	
	@Override
	public void open() throws CPSException {
		this.publisher = new AMQPClient(server, port, clientName, password);
		this.parser = MessageParserFactory.createMessageParser(SchemaFormat.json, parentSchema);
	}

	@Override
	public Collection<ParsedDataMessage> execute(ParsedDataMessage message)
			throws CPSException {
		ParsedDataMessage parsedMessage = (ParsedDataMessage) message;
		if (parsedMessage.getValueCount() != parentSchema.getTagCount()) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE,
					"The count of values( "
							+ Arrays.toString(parsedMessage.getObjectValues())
							+ ") should be equal to schema tag count ("
							+ Arrays.toString(parentSchema.getTagNames())
							+ ").");
		}

		Object[] values = parsedMessage.getObjectValues();
		if (values != null) {
			// String destination = (String) values[values.length - 1];
			publisher.publishMessage(topicName,
					parser.restoreFromDataMessage(parsedMessage));
		}
		return null;
	}

	@Override
	public void dispose() {
		publisher.close();
	}

	@Override
	public ITimeSeriesSchema getTimeSeriesSchema() {
		return null;
	}

}
